package server;


import concurrent.read.ReadThreadPool;
import concurrent.read.Reader;
import concurrent.write.WriteThreadPool;
import concurrent.write.Writer;
import http.Request;
import radio.Notifier;


import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by zzf on 2017/4/24.
 */
public class Server {


    private static Selector selector;
    private ServerSocketChannel serverChannel;
    private final int port;


    private Notifier notifier;

    private static AtomicInteger time = new AtomicInteger(0);

    private static volatile Queue<Future> futureQueue;

    private ExecutorService readerExecutor;

    private ExecutorService writerExecutor;

    private static final int THREAD_NUM = 100;

    public Server(int port) throws IOException {

        notifier = Notifier.getNotifier();

        this.port = port;
        //创建一个Selector实例(使用静态工厂方法open())
        selector = Selector.open();
        serverChannel = ServerSocketChannel.open();

        //设置成非阻塞
        serverChannel.configureBlocking(false);

        //绑定端口
        serverChannel.socket().bind(new InetSocketAddress(port));

        //绑定兴趣事件
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);

        readerExecutor = ReadThreadPool.newReadThreadPool(THREAD_NUM);

        writerExecutor = WriteThreadPool.newWriteThreadPool(THREAD_NUM);
        futureQueue = new ConcurrentLinkedQueue<>();
    }


    public void start() {
        System.out.println("服务端服务启动，监听端口" + port);

        while (true) {

            try {
                //该方法会阻塞等待，直到有一个或更多的信道准备好了I/O操作或等待超时。
                //select()方法将返回可进行I/O操作的信道数量。
                int num = selector.select();
                System.out.println("num:" + num);
                if (num > 0) {
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        SelectionKey key = iterator.next();
                        iterator.remove();
                        if (key.isAcceptable()) {
                            ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
                            SocketChannel socketChannel = serverChannel.accept();
                            socketChannel.configureBlocking(false);
                            notifier.notifierAccepted();
                            socketChannel.register(selector, SelectionKey.OP_READ);

                        } else if (key.isReadable()) {


                            //交给读线程池执行
                            Reader readerThread = new Reader(key);
//                            synchronized (futureQueue) {
                                futureQueue.add(readerExecutor.submit(readerThread));
//                            }

                            SocketChannel client = (SocketChannel) key.channel();
                            client.register(selector, SelectionKey.OP_WRITE);

                            System.out.println("提交读服务线程");

//                            time.incrementAndGet();
                        } else if (key.isWritable()) {
                            System.out.println("futureQueue.size:" + futureQueue.size());
                            //交给写线程池执行
                            if (!futureQueue.isEmpty()) {
//                                    while (!futureQueue.isEmpty()) {
                                Future<Request> future = null;
//                                synchronized (futureQueue) {
                                    future = futureQueue.remove();
//                                }

                                Writer writerThread = new Writer(key, future);
                                writerExecutor.execute(writerThread);
                                key.cancel();
//                                    }
                            } else {
                                key.cancel();
                            }


                            System.out.println("提交写服务线程向客户端发送回应数据");
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


}
